package simpledb.execution;

import simpledb.transaction.TransactionAbortedException;
import simpledb.common.DbException;
import simpledb.storage.Tuple;
import simpledb.storage.TupleDesc;

import java.util.*;

/**
 * The Join operator implements the relational join operation.
 */
public class Join extends Operator {

    private static final long serialVersionUID = 1L;

    private final JoinPredicate p;
    private OpIterator child1;
    private OpIterator child2;
    // 保存当前左表遍历到的节点，等待右表遍历完才调用左表的next
    private Tuple currentLeft;

    /**
     * Constructor. Accepts two children to join and the predicate to join them
     * on
     * 
     * @param p
     *            The predicate to use to join the children
     * @param child1
     *            Iterator for the left(outer) relation to join
     * @param child2
     *            Iterator for the right(inner) relation to join
     */
    public Join(JoinPredicate p, OpIterator child1, OpIterator child2) {
        // some code goes here
        this.p = p;
        this.child1 = child1;
        this.child2 = child2;
    }

    public JoinPredicate getJoinPredicate() {
        // some code goes here
        return p;
    }

    /**
     * @return
     *       the field name of join field1. Should be quantified by
     *       alias or table name.
     *       child1的getTupleDesc应该以及实现了别名，参考seqscan
     * */
    public String getJoinField1Name() {
        // some code goes here
        return child1.getTupleDesc().getFieldName(p.getField1());
    }

    /**
     * @return
     *       the field name of join field2. Should be quantified by
     *       alias or table name.
     * */
    public String getJoinField2Name() {
        // some code goes here
        return child2.getTupleDesc().getFieldName(p.getField1());
    }

    /**
     * @see TupleDesc#merge(TupleDesc, TupleDesc) for possible
     *      implementation logic.
     */
    public TupleDesc getTupleDesc() {
        // some code goes here
        return TupleDesc.merge(child1.getTupleDesc(), child2.getTupleDesc());
    }

    public void open() throws DbException, NoSuchElementException,
            TransactionAbortedException {
        // some code goes here
        super.open();
        child1.open();
        child2.open();
        currentLeft = null; // 初始化当前左表元组为null
    }

    public void close() {
        // some code goes here
        super.close();
        child1.close();
        child2.close();
        currentLeft = null; // 关闭时重置当前左表元组
    }

    public void rewind() throws DbException, TransactionAbortedException {
        // some code goes here
        super.close();
        super.open();
        child1.rewind();
        child2.rewind();
        currentLeft = null; // 回退时重置当前左表元组
    }

    /**
     * Returns the next tuple generated by the join, or null if there are no
     * more tuples. Logically, this is the next tuple in r1 cross r2 that
     * satisfies the join predicate. There are many possible implementations;
     * the simplest is a nested loops join.
     * <p>
     * Note that the tuples returned from this particular implementation of Join
     * are simply the concatenation of joining tuples from the left and right
     * relation. Therefore, if an equality predicate is used there will be two
     * copies of the join attribute in the results. (Removing such duplicate
     * columns can be done with an additional projection operator if needed.)
     * <p>
     * For example, if one tuple is {1,2,3} and the other tuple is {1,5,6},
     * joined on equality of the first column, then this returns {1,2,3,1,5,6}.
     * 
     * @return The next matching tuple.
     * @see JoinPredicate#filter
     */
    protected Tuple fetchNext() throws TransactionAbortedException, DbException {
        while (true) {
            // 第一次调用，还未初始化, 或者右表已经遍历完需要下一个了。
            if (currentLeft==null) {
                // 左表遍历完了或者本身为空
                if (!child1.hasNext()) {
                    return null;
                }
                currentLeft = child1.next();
            }
            while (child2.hasNext()) {
                Tuple rightTuple = child2.next();
                // 检查是否满足连接条件
                if (p.filter(currentLeft, rightTuple)) {
                    // 合并两个元组（左表元组字段在前，右表元组字段在后）
                    Tuple mergedTuple = new Tuple(getTupleDesc());
                    int fieldIndex = 0;

                    // 添加左表元组的所有字段
                    for (int i = 0; i < currentLeft.getTupleDesc().numFields(); i++) {
                        mergedTuple.setField(fieldIndex++, currentLeft.getField(i));
                    }

                    // 添加右表元组的所有字段
                    for (int i = 0; i < rightTuple.getTupleDesc().numFields(); i++) {
                        mergedTuple.setField(fieldIndex++, rightTuple.getField(i));
                    }

                    return mergedTuple;
                }
            }

            // 右表已遍历完毕，重置右表并准备处理左表的下一个元组
            child2.rewind();
            currentLeft = null;
        }
    }


    @Override
    public OpIterator[] getChildren() {
        // some code goes here
        return new OpIterator[] {child1, child2};
    }

    @Override
    public void setChildren(OpIterator[] children) {
        // some code goes here
        child1 = children[0];
        child2 = children[1];
    }

}
